/*
 *
 * Copyright (c) 2020 The Raptor Authors. All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

#include "src/linux/connection.h"

#include <errno.h>
#include <string.h>
#include <sys/epoll.h>
#include <sys/socket.h>

#include "raptor-lite/impl/handler.h"
#include "raptor-lite/impl/endpoint.h"
#include "raptor-lite/utils/log.h"

#include "src/common/common.h"
#include "src/common/endpoint_impl.h"

#include "src/linux/epoll_thread.h"
#include "src/linux/socket_settings.h"

namespace raptor {

Connection::Connection(std::shared_ptr<EndpointImpl> obj)
    : service_(nullptr)
    , proto_(nullptr)
    , recv_thread_(nullptr)
    , send_thread_(nullptr) {

    next_package_size_ = 0;
    endpoint_ = obj;
}

Connection::~Connection() { Shutdown(false, nullptr); }

void Connection::Init(int32_t index) { index_ = index; }

void Connection::SetPollingThread(PollingThread *recv_thread, PollingThread *send_thread) {
    recv_thread_ = recv_thread;
    send_thread_ = send_thread;

    recv_thread_->Add(static_cast<int>(endpoint_->fd_),
                      reinterpret_cast<void *>(endpoint_->connection_id_), EPOLLIN | EPOLLET);
    send_thread_->Add(static_cast<int>(endpoint_->fd_),
                      reinterpret_cast<void *>(endpoint_->connection_id_), EPOLLOUT | EPOLLET);
}
void Connection::SetTransferService(internal::NotificationTransferService *service) {
    service_ = service;
}

void Connection::SetProtocol(ProtocolHandler *p) { proto_ = p; }

bool Connection::SendMsg(const Slice &s) {
    if (!endpoint_->IsValid()) return false;
    // AutoMutex g(&_snd_mutex);
    snd_buffer_.AddSlice(s);
    send_thread_->Modify(static_cast<int>(endpoint_->fd_),
                         reinterpret_cast<void *>(endpoint_->connection_id_), EPOLLOUT | EPOLLET);
    return true;
}

void Connection::Shutdown(bool notify, raptor_error desc) {
    if (!endpoint_->IsValid()) {
        return;
    }

    recv_thread_->Delete(static_cast<int>(endpoint_->fd_), EPOLLIN | EPOLLET);
    send_thread_->Delete(static_cast<int>(endpoint_->fd_), EPOLLOUT | EPOLLET);
    int fd = static_cast<int>(endpoint_->fd_);

    raptor_set_socket_shutdown(static_cast<int>(endpoint_->fd_));
    endpoint_->fd_ = uint64_t(~0);

    if (notify) {
        service_->OnClosed(endpoint_, desc);
    }

    //_rcv_mutex.Lock();
    if (!rcv_buffer_.Empty()) {
        log_error("Fd:%d have %lu bytes in rcv_buffer not been processed and will be discarded", fd,
                  rcv_buffer_.GetBufferLength());
    }
    rcv_buffer_.ClearBuffer();
    //_rcv_mutex.Unlock();

    //_snd_mutex.Lock();
    if (!snd_buffer_.Empty()) {
        log_error("Fd:%d have %lu bytes in snd_buffer not been processed and will be discarded", fd,
                  snd_buffer_.GetBufferLength());
    }
    snd_buffer_.ClearBuffer();
    //_snd_mutex.Unlock();
}

bool Connection::IsOnline() { return endpoint_->IsValid(); }

void Connection::DoHeartBeat() { service_->OnHeartBeat(endpoint_); }

raptor_error Connection::DoRecvEvent(EventDetail *detail) {
    int result = OnRecv();
    if (result == 0) {
        recv_thread_->Modify(static_cast<int>(endpoint_->fd_),
                             reinterpret_cast<void *>(endpoint_->connection_id_),
                             EPOLLIN | EPOLLET);
        return RAPTOR_ERROR_NONE;
    }
    return RAPTOR_POSIX_ERROR("Connection:OnRecv");
}

raptor_error Connection::DoSendEvent(EventDetail *detail) {
    int result = OnSend();
    if (result == -1) {
        send_thread_->Modify(static_cast<int>(endpoint_->fd_),
                             reinterpret_cast<void *>(endpoint_->connection_id_),
                             EPOLLOUT | EPOLLET);
    }
    return RAPTOR_ERROR_NONE;
}

int Connection::OnRecv() {
    // AutoMutex g(&_rcv_mutex);

    int recv_bytes = 0;
    int unused_space = 0;
    do {
        char buffer[8192];

        unused_space = sizeof(buffer);
        recv_bytes = ::recv(static_cast<int>(endpoint_->fd_), buffer, unused_space, 0);

        if (recv_bytes == 0) {
            return -1;
        }

        if (recv_bytes < 0) {
            if (errno == EINTR || errno == EWOULDBLOCK || errno == EAGAIN) {
                return 0;
            }
            return -1;
        }

        if (!proto_) {
            service_->OnDataReceived(endpoint_, Slice(buffer, recv_bytes));
        } else {
            // Add to recv buffer
            rcv_buffer_.AddSlice(Slice(buffer, recv_bytes));
            if (rcv_buffer_.GetBufferLength() >= next_package_size_) {
                if (ParsingProtocol() == -1) {
                    return -1;
                }
            }
        }
    } while (recv_bytes == unused_space);
    return 0;
}

int Connection::OnSend() {
    // AutoMutex g(&_snd_mutex);
    if (snd_buffer_.Empty()) {
        return 0;
    }

    size_t count = 0;
    do {

        Slice slice = snd_buffer_.Front();
        int slen = ::send(static_cast<int>(endpoint_->fd_), slice.begin(), slice.size(), 0);

        if (slen == 0) {
            break;
        }

        if (slen < 0) {
            if (errno == EINTR || errno == EWOULDBLOCK || errno == EAGAIN) {
                return -1;
            }
            break;
        }
        snd_buffer_.MoveHeader((size_t)slen);
        count = snd_buffer_.SliceCount();

    } while (count > 0);
    return 0;
}

bool Connection::ReadSliceFromRecvBuffer(size_t read_size, Slice &s) {
    size_t cache_size = rcv_buffer_.GetBufferLength();
    if (read_size >= cache_size) {
        s = rcv_buffer_.Merge();
        return true;
    }
    s = rcv_buffer_.GetHeader(read_size);
    return false;
}

int Connection::ParsingProtocol() {
    constexpr size_t header_size = 256;

    size_t cache_size = rcv_buffer_.GetBufferLength();
    int package_counter = 0;

    if (cache_size >= next_package_size_ && next_package_size_ > 0) {
        Slice package = rcv_buffer_.GetHeader(next_package_size_);
        service_->OnDataReceived(endpoint_, package);
        rcv_buffer_.MoveHeader(next_package_size_);
        cache_size -= next_package_size_;
        next_package_size_ = 0;
    }

    while (cache_size > 0) {
        size_t read_size = header_size;
        int pack_len = 0;
        Slice package;
        do {
            bool reach_tail = ReadSliceFromRecvBuffer(read_size, package);
            pack_len = proto_->OnCheckPackageLength(endpoint_, package.begin(), package.size());
            if (pack_len < 0) {
                log_warn("Connection: Internal protocol parsing error");
                return -1;
            }

            // equal 0 means we need more data
            if (pack_len == 0) {
                if (reach_tail) {
                    goto done;
                }
                read_size *= 2;
                continue;
            }

            // We got a complete packet with length pack_len
            if (cache_size >= static_cast<size_t>(pack_len)) {
                break;
            }
            next_package_size_ = pack_len;
            goto done;
        } while (true);

        if (package.size() < static_cast<size_t>(pack_len)) {
            package = rcv_buffer_.GetHeader(pack_len);
        } else {
            size_t n = package.size() - pack_len;
            package.RemoveTail(n);
        }
        service_->OnDataReceived(endpoint_, package);
        rcv_buffer_.MoveHeader(pack_len);

        cache_size = rcv_buffer_.GetBufferLength();
        package_counter++;
    }
done:
    return package_counter;
}

} // namespace raptor
